package com.lagou.rpc.server.zookeeper;

import com.alibaba.fastjson.JSONObject;
import com.lagou.rpc.common.SimpleRpcConstants;
import com.lagou.rpc.common.entity.HostPortMetaData;
import com.lagou.rpc.common.exceptions.SimpleRpcException;
import com.lagou.rpc.server.SimpleRpcMethod;
import com.lagou.rpc.server.SimpleRpcMethodRegistry;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ZookeeperRpcMethodRegistry
 *
 * @author wlz
 * @date 2020/5/20
 */
public class ZookeeperRpcMethodRegistry extends SimpleRpcMethodRegistry {

    private static final Logger logger = LoggerFactory.getLogger(ZookeeperRpcMethodRegistry.class);

    private RefreshResponseTimeScheduler refreshResponseTimeScheduler;

    public ZookeeperRpcMethodRegistry(ZookeeperRegistryConfig zookeeperRegistryConfig) {
        String serverName = zookeeperRegistryConfig.getServerName();
        HostPortMetaData hostPortMetaData = zookeeperRegistryConfig.getHostPortMetaData();
        String zooServerAddress = zookeeperRegistryConfig.getZooServerAddress();
        String namespace = zookeeperRegistryConfig.getNamespace();
        // 创建zk客户端
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(zooServerAddress)
                .retryPolicy(retryPolicy)
                // 包含隔离名称
                .namespace(namespace)
                .build();
        client.start();
        logger.info("成功连接zookeeper " + zooServerAddress);
        // 注册服务节点
        try {
            String nodeName = SimpleRpcConstants.SERVER_HOST_LIST_NODE + "/" + serverName;
            Stat stat = client.checkExists().forPath(nodeName);
            if (stat != null) {
                client.delete().forPath(nodeName);
            }
            client.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL)
                    .forPath(nodeName, JSONObject.toJSONBytes(hostPortMetaData));
            logger.info("成功注册节点" + nodeName);
        } catch (Exception e) {
            throw new SimpleRpcException("服务注册失败", e);
        }
        // 创建刷新响应时间的定时任务
        refreshResponseTimeScheduler = new RefreshResponseTimeScheduler(hostPortMetaData, serverName, client);
        refreshResponseTimeScheduler.startRefreshTask(5);
    }

    @Override
    public SimpleRpcMethod getRpcMethod(String methodId) {
        // 封装SimpleRpcMethod,返回ZookeeperRpcMethod
        SimpleRpcMethod simpleRpcMethod = super.getRpcMethod(methodId);
        if (simpleRpcMethod == null) {
            throw new SimpleRpcException("no method found");
        }
        return new ZooKeeperRpcMethod(simpleRpcMethod, refreshResponseTimeScheduler);
    }
}
